package com.mes.cloud.commons;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 超时缓存工具类
 * @param <K>
 * @param <V>
 */
public class TimeoutCache<K, V> {

    private static final Logger LOG = LoggerFactory.getLogger(TimeoutCache.class.getName());

    private ConcurrentMap<K, V> cacheMap = new ConcurrentHashMap<K, V>();

    private DelayQueue<DelayItem<K>> queue = new DelayQueue<>();

    private Thread daemonThread;


    public interface Callback<K, V> {
        void onTimeout(K key, V value);
    }

    private static class DelayItem<T> implements Delayed {
        private static final long NANO_ORIGIN = System.nanoTime();

        final static long now() {
            return System.nanoTime() - NANO_ORIGIN;
        }

        private static final AtomicLong sequencer = new AtomicLong(0);

        private final long sequenceNumber;

        private final long time;

        private final T item;

        public DelayItem(T submit, long timeout) {
            this.time = now() + timeout;
            this.item = submit;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        public T getItem() {
            return this.item;
        }

        public long getDelay(TimeUnit unit) {
            long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);
            return d;
        }

        public int compareTo(Delayed other) {
            if (other == this)
                return 0;
            if (other instanceof DelayItem) {
                DelayItem x = (DelayItem) other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }

        @Override
        public int hashCode() {
            return item.hashCode();
        }

        @Override
        public boolean equals(Object other) {
            if (other == this)
                return true;
            if (other instanceof DelayItem) {
                DelayItem x = (DelayItem) other;
                return item.equals(x.item);
            }
            return false;
        }

    }

    public TimeoutCache() {
        this(null);
    }

    public TimeoutCache(Callback<K, V> callback) {
        Runnable daemonTask = () -> daemonCheck(callback);

        daemonThread = new Thread(daemonTask);
        daemonThread.setDaemon(true);
        daemonThread.setName("TimeoutCache Daemon");
        daemonThread.start();
    }

    public String toString() {
        StringJoiner stringJoiner = new StringJoiner(";");
        Iterator iterator = cacheMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry entry = (Map.Entry) iterator.next();
            Object key = entry.getKey();
            Object val = entry.getValue();
            stringJoiner.add(String.format("%s:%s",key,val));
        }
        return stringJoiner.toString();
    }

    private void daemonCheck(Callback<K, V> callback) {

        if (LOG.isInfoEnabled())
            LOG.debug("Timeout Cache check thread started.");

        while (true) {
            try {
                DelayItem<K> delayItem = queue.take();
                if (delayItem != null && callback != null) {
                    K key = delayItem.getItem();
                    if (key == null) continue;

                    V value = cacheMap.remove(key);
                    if (value != null) {
                        callback.onTimeout(key, value);
                    }
                }
            } catch (InterruptedException e) {
                if (LOG.isErrorEnabled())
                    LOG.error(e.getMessage(), e);
                break;
            }
        }

        if (LOG.isInfoEnabled())
            LOG.debug("Timeout Cache check thread stopped.");
    }

    public void put(K key, V value, long time, TimeUnit unit) {
        long nanoTime = TimeUnit.NANOSECONDS.convert(time, unit);
        DelayItem<K> newItem = new DelayItem<>(key, nanoTime);

        V oldItem = cacheMap.put(key, value);
        if (oldItem != null)
            queue.remove(newItem);

        queue.put(newItem);
    }

    public V replace(K key, V value) {
        return cacheMap.put(key, value);
    }

    public V remove(K key) {
        V oldItem = cacheMap.remove(key);
        if(oldItem!=null) {
            List<DelayItem> list = queue.stream().filter(item->item.getItem().equals(key)).collect(Collectors.toList());
            queue.removeAll(list);
        }

        return oldItem;
    }

    public boolean containsKey(K key) {
        return cacheMap.containsKey(key);
    }

    public void clear() {
        cacheMap.clear();
        queue.clear();
    }

    public V get(K key) {
        return cacheMap.get(key);
    }

    public Set<K> keys() {
        return cacheMap.keySet();
    }
}
